package org.cancer.service


import org.apache.spark.sql.{DataFrame, SaveMode}
import org.apache.spark.sql.functions._
import org.cancer.util.SparkUtil

class CancerStageSurvivalService_Faye {
  def CancerStageSurvival(data: DataFrame): Unit = {
    val spark = SparkUtil.takeSpark()
    import spark.implicits._

    // 数据预处理和转换
    val processedData = data
      .withColumn("癌症分期", $"癌症分期".cast("string"))
      .withColumn("生存状态", when($"生存状态" === "已故", "Deceased")
        .otherwise(when($"生存状态" === "在世", "Alive").otherwise($"生存状态")))
      .filter($"癌症分期".isNotNull && $"生存状态".isNotNull)

    // 计算每个癌症分期的患者总数
    val totalByStage = processedData
      .groupBy("癌症分期")
      .agg(count("*").alias("总患者数"))

    // 计算每个癌症分期的存活患者数
    val survivedByStage = processedData
      .filter($"生存状态" === "Alive")
      .groupBy("癌症分期")
      .agg(count("*").alias("存活患者数"))

    // 计算存活率（存活患者数/总患者数）
    val survivalRateByStage = totalByStage
      .join(survivedByStage, Seq("癌症分期"), "left_outer")
      .na.fill(0, Seq("存活患者数")) // 将NULL值替换为0
      .withColumn("存活率", round($"存活患者数" / $"总患者数" * 100, 2))
      .select("癌症分期", "总患者数", "存活患者数", "存活率")
      .orderBy("癌症分期")

    survivalRateByStage.show()

    survivalRateByStage.write
      .format("jdbc") //指定数据源 链接mysql odbc  ORACLE
      .option("url", "jdbc:mysql://node1:3306/cancer_patients") //链接本地 jdbc:mysql://localhost:3306/ 本地的数据库
      .option("driver", "com.mysql.cj.jdbc.Driver") //mysql 8.0以上：com.mysql.cj.jdbc.Driver
      .option("user", "root") //用户名
      .option("password", "123456") //密码
      .option("dbtable", "cancer_stage_survival_faye") //表名
      .mode(SaveMode.Append) //当数据不存在的时候会自动创建
      .save()

    spark.stop()
  }
}
